package org.atguigu.gmall.realtime.join;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * @Author zenghaifeng
 * @Date 2023/4/25 0:26
 * @Package: org.atguigu.gmall.realtime.join
 * @Class: LeftJoinKafka
 * @Description: TODO
 * @Version 1.0
 */
public class LeftJoinKafka extends BaseSQLApp {
    public static void main(String[] args) {
        new LeftJoinKafka().init(50000, 2, "LeftJoinKafka");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        tbe.executeSql("create table a(" +
                " id string, " +
                " name string, " +
                " c int " +
                ")" + SQLUtil.getKafkaSourceDDL("a", "LeftJoinKafka1", "csv"));

        tbe.executeSql("create table b(" +
                " id string, " +
                " age int " +
                ")" + SQLUtil.getKafkaSourceDDL("b", "LeftJoinKafka2", "csv"));


        Table result = tbe.sqlQuery("select " +
                " a.id id," +
                " name, " +
                " c, " +
                " age " +
                "from a " +
                "left join b on a.id=b.id");

        tbe.executeSql("create table ab(" +
                "  id string, " +
                "  name string, " +
                "  c int, " +
                "  age int," +
                "  primary key(id) not enforced " +
                ")" + SQLUtil.getUpsertKafkaDDL("ab_left_1", "json"));

        result.executeInsert("ab");
    }
}
